package com.smile.source;

import com.smile.beans.SensorReading;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Properties;
import java.util.Random;

public class ArraySource {
    public static void main(String[] args) throws Exception {
//        //1.从集合中读取数据
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        //获取数据源
//        DataStreamSource<SensorReading> streamSource = env.fromCollection(
//                Arrays.asList(
//                        new SensorReading("tom1",18,"xian"),
//                        new SensorReading("tom2",20,"xian"),
//                        new SensorReading("tom3",22,"xian")
//        ));
//        //打印
//        streamSource.print();
//        //执行
//        env.execute();
//
//        //2.从文件中读取数据
//        //DataStreamSource<String> fileSource = env.readTextFile("YOUR_FILE_PATH");
//
//        //3.从kafka中读取
//        // kafka 配置项
//        Properties properties = new Properties();
//        properties.setProperty("bootstrap.servers", "localhost:9092");
//        properties.setProperty("group.id", "consumer-group");
//        properties.setProperty("key.deserializer",
//                "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.setProperty("value.deserializer",
//                "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.setProperty("auto.offset.reset", "latest");
//        // 从 kafka 读取数据
//        DataStream<String> dataStream = env.addSource( new
//                FlinkKafkaConsumer011<String>("sensor", new SimpleStringSchema(), properties));

        //4.自定义source
        DataStream<SensorReading> dataStream2 = env.addSource( new MySensor());

     }
    public static class MySensor implements SourceFunction<SensorReading> {
        private boolean running = true;
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            Random random = new Random();
            HashMap<String, Double> sensorTempMap = new HashMap<String, Double>();
            for( int i = 0; i < 10; i++ ){
                sensorTempMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
            }
            while (running) {
                for( String sensorId: sensorTempMap.keySet() ){
                    Double newTemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                    sensorTempMap.put(sensorId, newTemp);
                    ctx.collect( new SensorReading(sensorId, System.currentTimeMillis(),
                            newTemp));
                }
                Thread.sleep(1000L);
            }
        }
        public void cancel() {
            this.running = false;
        }
    }

}
